package com.example.redis.util;

import cn.hutool.core.util.StrUtil;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 延迟队列工具
 *
 * @Author guoyb
 * @Date 2022/6/28  11:29
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisDelayQueueUtil {

    private final RedissonClient redissonClient;


    //由于延时队列持久化在redis中，所以机器宕机数据不会异常丢失，机器重启后，会正常消费队列中积累的任务。


// ：延时队列的过期机制是靠客户端的监听（定时任务+redis订阅）实现的，所以在消费数据的时候要先开启监听，
//             监听的开启逻辑在redissonClient.getDelayedQueue()里面（会触发一次任务执行，并开启定时任务）
    /**
     * 添加延迟队列
     *
     * @param value     队列值
     * @param delay     延迟时间
     * @param timeUnit  时间单位
     * @param queueCode 队列键
     * @param <T>
     */
    public <T> boolean addDelayQueue(@NonNull T value, @NonNull long delay, @NonNull TimeUnit timeUnit, @NonNull String queueCode) {
        if (StrUtil.isBlank(queueCode) || Objects.isNull(value)) {
            return false;
        }
        try {
            //生成阻塞队列的语句是
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);

            //生成延迟队列的语句是
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            delayedQueue.offer(value, delay, timeUnit);

            // 异步操作
            // delayedQueue.offerAsync("task: ", delay, timeUnit);

            //delayedQueue.destroy();
            log.info("(添加延时队列成功) 队列键：{}，队列值：{}，延迟时间：{}", queueCode, value, timeUnit.toSeconds(delay) + "秒");
        } catch (Exception e) {
            log.error("(添加延时队列失败) {}", e.getMessage());
            throw new RuntimeException("(添加延时队列失败)");
        }
        return true;
    }



    // peek：获取队列的head对象，但不是从队列中移除。如果队列空，返回空
    // poll ：​获取并移出队列head对象，如果head没有超时，返回空
    // take ：​获取并移出队列head对象，如果没有超时head对象，会wait当前线程直到有对象满足超时条件

    /**
     * 获取延迟队列
     *
     * @param queueCode
     * @param <T>
     */
    public <T> T getDelayQueue(@NonNull String queueCode) throws InterruptedException {
        if (StrUtil.isBlank(queueCode)) {
            return null;
        }
        RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueCode);
        RDelayedQueue<Map> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);

        //阻塞队列有数据就返回，否则wait
        // 该方法会阻塞，直到能获取到任务
        // blockingDeque.take();

        T value = (T) blockingDeque.poll();
        return value;
    }


    // 客户端启动，redisson先订阅一个key，同时 BLPOP key 0 无限监听一个阻塞队列（等里面有数据了就返回）。
    // 当有数据put时，redisson先把数据放到一个zset集合（按延时到期时间的时间戳为分数排序），同时发布上面订阅的key，发布内容为数据到期的timeout，此时客户端进程开启一个延时任务，延时时间为发布的timeout。
    // 客户端进程的延时任务到了时间执行，从zset分页取出过了当前时间的数据，然后将数据rpush到第一步的阻塞队列里。然后将当前数据从zset移除，取完之后，又执行 BLPOP key 0 无限监听一个阻塞队列。
    // 上一步客户端监听的阻塞队列返回取到数据，回调到 RBlockingQueue 的 take方法。于是，我们就收到了数据。


    /**
     * 删除指定队列中的消息
     *
     * @param o         指定删除的消息对象队列值(同队列需保证唯一性)
     * @param queueCode 指定队列键
     */
    public boolean removeDelayedQueue(@NonNull Object o, @NonNull String queueCode) {
        if (StrUtil.isBlank(queueCode) || Objects.isNull(o)) {
            return false;
        }
        RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        return delayedQueue.remove(o);
    }

}
